package www.bwsensing.com.common.netty.handler;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.socket.DatagramPacket;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;
import www.bwsensing.com.common.cache.RedisService;
import www.bwsensing.com.common.mqtt.MqttPushClient;
import www.bwsensing.com.common.thread.NamedThreadFactory;
import www.bwsensing.com.common.utills.StringUtils;
import www.bwsensing.com.service.collection.database.dataobject.ServiceData;
import www.bwsensing.com.stream.DataStreamReceive;
import www.bwsensing.com.stream.object.DataMessage;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author macos-zyj
 */
@Slf4j
@ChannelHandler.Sharable
public class NettyUdpServerHandler extends SimpleChannelInboundHandler<DatagramPacket>
        implements DataPushHandle{
    /**
     * 服务配置信息
     */
    private final ServiceData serviceData;

    private static final Long WAIT_TIMEOUT = 20000L;

    private final ExecutorService nettyUdpThread = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors() + 1,
            Runtime.getRuntime().availableProcessors() + 1,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1000),
            new NamedThreadFactory("ASYNC-UDP-CACHE-POOL"));

    public static NettyUdpServerHandler nettyUdpServerHandler;

    private static final ConcurrentHashMap<String, List<DataMessage>>
            IP_DATA_CACHE = new ConcurrentHashMap<>();

    private final RedisService redisService;

    private final MqttPushClient mqttPushClient;

    public NettyUdpServerHandler(ServiceData serviceData,
                                 RedisService redisService, MqttPushClient mqttPushClient) {
        this.serviceData = serviceData;
        this.redisService = redisService;
        this.mqttPushClient = mqttPushClient;
    }

    @PostConstruct
    public void init(){
        nettyUdpServerHandler = this;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, DatagramPacket msg) {
        InetSocketAddress  remoteAddress = msg.sender();
        String address = remoteAddress.getAddress().getHostAddress();
        String message = msg.content().toString(CharsetUtil.UTF_8);
        if (StringUtils.isNotEmpty(message)){
            log.info(message);
            storageAndWait(address,message);
        }
    }

    private  void storageAndWait(String ipAddress,String receiveMessage){
        long currentTime = System.currentTimeMillis();
        Timestamp timestamp = new Timestamp(currentTime);
        receiveMessage = receiveMessage.replace("\r","").replace(" ","");
        if (redisService.hasKey(ipAddress)&& null != IP_DATA_CACHE.get(ipAddress)){
            IP_DATA_CACHE.get(ipAddress).add(new DataMessage(timestamp,receiveMessage));
            redisService.setCacheObject(ipAddress,receiveMessage,WAIT_TIMEOUT,TimeUnit.MILLISECONDS);
        } else {
            List<DataMessage> messages = new ArrayList<>();
            messages.add(new DataMessage(timestamp,receiveMessage));
            IP_DATA_CACHE.put(ipAddress,messages);
            redisService.setCacheObject(ipAddress,receiveMessage,WAIT_TIMEOUT,TimeUnit.MILLISECONDS);
            commitThreadWaitBatch(ipAddress);
        }
    }

    private void commitThreadWaitBatch(String ipAddress){
        nettyUdpThread.submit(() -> {
            while(true){
                if(!redisService.hasKey(ipAddress)){
                    if (IP_DATA_CACHE.get(ipAddress).size()>0){
                        DataStreamReceive receiveData = getDataStreamPush(IP_DATA_CACHE.get(ipAddress),"UDP",serviceData);
                        mqttPushClient.publishEvent(receiveData);
                        log.info(receiveData.getMessage());
                        log.info("facility data pushed,Client address:{}",ipAddress);
                    }
                    IP_DATA_CACHE.remove(ipAddress);
                    log.info("link down ,Client address:{}",ipAddress);
                    break;
                } else {
                    try {
                        Thread.sleep(WAIT_TIMEOUT/4);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
    }

}
